package com.aydx.minirpc.core.register;


import com.aydx.minirpc.core.MiniRpcException;
import com.aydx.minirpc.core.proxy.NodeInfo;
import com.aydx.minirpc.core.common.JsonUtils;
import com.aydx.minirpc.core.common.ZooKeeperUtils;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * Created by aydx on 2019/1/6.
 */
public class ZookeeperRegister extends RpcRegister implements Watcher{

    private Logger logger = LoggerFactory.getLogger(ZookeeperRegister.class);
    private ZooKeeper zooKeeper;

    private static CountDownLatch latch=new CountDownLatch(1);

    private ZookeeperRegister(String name,String registerCenter,int timeOut) throws IOException {
        this.zooKeeper= new ZooKeeper(registerCenter,timeOut,this);
        setName(name);
    }
    public static ZookeeperRegister newInstance(String name,String registerCenter,int timeOut) throws IOException, InterruptedException {
        ZookeeperRegister zookeeperRegister=new ZookeeperRegister(name,registerCenter,timeOut);
        latch.await(timeOut, TimeUnit.SECONDS);
        return zookeeperRegister;
    }

    public static void main(String[] args) throws Exception {
        ZookeeperRegister register=ZookeeperRegister.newInstance("default","127.0.0.1:2181",5000000);
        NodeInfo nodeInfo=new NodeInfo().setNodeType(NodeType.PROVIDER).setGroup("test").setVersion("1.0").setInterfaceName("echoService").setIp("127.0.0.1").setPort(2281);
        register.register(nodeInfo);
        //
        List<NodeInfo> list=register.getNode(NodeType.PROVIDER,nodeInfo.getInterfaceName(),nodeInfo.getGroup(),nodeInfo.getVersion());
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        if(Event.KeeperState.SyncConnected==watchedEvent.getState()){
            logger.info("zookeeper 链接成功",watchedEvent.getType().getIntValue());
            latch.countDown();
        }
        logger.info("#ZookeeperRegister event",watchedEvent.getType().getIntValue());
    }

    @Override
    public void register(NodeInfo nodeInfo) throws Exception {
        String uri=UriGenerator.gen(nodeInfo);
        ZooKeeperUtils.createZNode(uri,"",zooKeeper, CreateMode.PERSISTENT);
        //
        uri=uri+"/"+nodeInfo.getIp()+","+nodeInfo.getPort();
        ZooKeeperUtils.createZNode(uri,JsonUtils.toJson(nodeInfo),zooKeeper,CreateMode.EPHEMERAL);
    }

    @Override
    public List<NodeInfo> getNode(NodeType nodeType,String interfaceName,String group,String version) throws InterruptedException, KeeperException {
        NodeInfo nodeInfo=new NodeInfo().setInterfaceName(interfaceName).setGroup(group).setVersion(version).setNodeType(nodeType);
        String uri=UriGenerator.gen(nodeInfo);
        if(this.zooKeeper.exists(uri,false)==null){
            throw new MiniRpcException("没有提供者"+interfaceName);
        }
        List<String> childs=this.zooKeeper.getChildren(uri,false);
        if(childs==null || childs.size()==0){
            return null;
        }else{
            List<NodeInfo> nodeInfos=new ArrayList<>();
            for(String child:childs){
                String path=uri+"/"+child;
                Stat stat=new Stat();
                byte[] bytes=this.zooKeeper.getData(path,this,stat);
                if(bytes!=null&& bytes.length>0){
                    NodeInfo nodeInfoTemp=JsonUtils.fromJson(new String(bytes),NodeInfo.class);
                    nodeInfos.add(nodeInfoTemp);
                }
            }
            return nodeInfos;
        }
    }
}
